package com.alison.sink
import java.util.{Timer, TimerTask}
import org.apache.spark.sql.SparkSession
object E1_std_memory {


  def main(args: Array[String]): Unit = {
    memory_out
  }

  def console_out()={
    // 创建 SparkSession
    val spark = SparkSession.builder()
      .appName("HelloStructuredStreaming")
      .master("local[*]")
      .getOrCreate()

    // 导入隐式转换
    import spark.implicits._

    // 创建一个流式DataFrame，这里从Socket读取数据
    val lines = spark.readStream
      .format("socket")
      .option("host", "192.168.56.104")
      .option("port", 9999)
      .load()

    // 对流式DataFrame进行转换，将每一行数据切分成单词
    val words = lines.as[String].flatMap(_.split(" "))

    // 对单词进行分组和聚合
    val wordCounts = words.groupBy("value").count()

    // 启动查询, 把结果打印到控制台
    val query = wordCounts.writeStream
      .outputMode("complete") // 或者选择其他适合的输出模式
      .format("console") // 输出至控制台
      .start()
    // 等待应用程序终止
    query.awaitTermination()
    //关闭 Spark
    spark.stop()
  }
  """
    |1，基本介绍
    |（1）memory sink 也是用于测试，其将统计结果全部输入内存中指定的表中，然后可以通过 sql 与从表中查询数据。
    |注意：如果数据量非常大, 可能会发送内存溢出。
    |
    |（2）memory sink 支持 Append、Complete 这几种输出模式。
    |
    |
    |操作：
    |1. nc -lk 9999
    |java go
    |flink spark
    |
    |""".stripMargin
  def memory_out()={
    // 创建 SparkSession
    val spark = SparkSession.builder()
      .appName("HelloStructuredStreaming")
      .master("local[*]")
      .getOrCreate()

    // 导入隐式转换
    import spark.implicits._

    // 创建一个流式DataFrame，这里从Socket读取数据
    val lines = spark.readStream
      .format("socket")
      .option("host", "192.168.56.104")
      .option("port", 9999)
      .load()

    // 对流式DataFrame进行转换，将每一行数据切分成单词
    val words = lines.as[String].flatMap(_.split(" "))

    // 对单词进行分组和聚合
    val wordCounts = words.groupBy("value").count()

    // 启动查询, 把输出至内存
    val query = wordCounts.writeStream
      .outputMode("complete") // 或者选择其他适合的输出模式
      .format("memory") // 输出至内存
      .queryName("word_count") // 内存临时表名
      .start()

    // 测试使用定时器执行查询表
    val timer = new Timer(true)
    val task: TimerTask = new TimerTask {
      override def run(): Unit = spark.sql("select * from word_count").show
    }
    timer.scheduleAtFixedRate(task, 0, 10000)
    // 等待应用程序终止
    query.awaitTermination()
    //关闭 Spark
    spark.stop()
  }
}
